package test;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;

import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Random;

/**
 * 模拟数据程序
 *
 * @author Administrator
 */
public class MockData {

    public static void mockData(SparkSession sparkSession) {
        mockProductInfo(sparkSession);
        mockUserInfo(sparkSession);
        mockVisitAction(sparkSession);
    }

    public static void mockProductInfo(SparkSession sparkSession) {
        ArrayList<Row> productList = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            String pId = null;
            String pName = null;
            String categoryId = null;
            categoryId = "category" + i;
            for (int j = 1; j <= 10; j++) {
                pId = "pid" + j + "_" + i;
                pName = "pname" + j + "_" + i;
                productList.add(RowFactory.create(pId, pName, categoryId));
            }
        }
      //不要采用映射的方法，因为映射的方法字段顺序不确定（是按照字典顺序的）
        StructType schema = DataTypes.createStructType(Arrays.asList(
                DataTypes.createStructField("pId", DataTypes.StringType,true),
                DataTypes.createStructField("pName", DataTypes.StringType,true),
                DataTypes.createStructField("categoryId", DataTypes.StringType,true)

        ));
        Dataset<Row> productDS = sparkSession.createDataFrame(productList, schema);
        productDS.createOrReplaceTempView("product_info");
       // productDS.show();
    }
    /*表名：user_info（Hive表）
        user_id：其实就是每一个用户的唯一标识，通常是自增长的Long类型，BigInt类型
        name：每个用户自己的昵称、或者是真实姓名
        sex:性别
        age:年龄*/
    public static void mockUserInfo(SparkSession sparkSession){
        Random random = new Random();
        ArrayList<Row> userList = new ArrayList<>();
        for (int i = 1; i <= 40; i++) {
            String userId = null;
            String name = null;
            String sex = null;
            int age = random.nextInt(50)+10;
            userId = "userid"+i;
            name = "name"+i;
            sex = (i%2==0) ? "man" : "woman";
            userList.add(RowFactory.create(userId, name, sex, age));
        }

        StructType schema = DataTypes.createStructType(Arrays.asList(
                DataTypes.createStructField("userId", DataTypes.StringType,true),
                DataTypes.createStructField("name", DataTypes.StringType,true),
                DataTypes.createStructField("sex", DataTypes.StringType,true),
                DataTypes.createStructField("age", DataTypes.IntegerType,true)
        ));
        Dataset<Row> userDS = sparkSession.createDataFrame(userList, schema);
        userDS.createOrReplaceTempView("user_info");
        //44| name1|woman| userid1
        //userDS.show();
    }

    public static void mockVisitAction(SparkSession sparkSession){
        JavaRDD<String> lineJavaRDD = sparkSession.read().textFile("C:\\Users\\79293\\Desktop\\找实习加油\\sparkDemo01\\原始数据\\user_visit_action.txt")
                .toJavaRDD();

        JavaRDD<String> filterRDD = lineJavaRDD.filter(line -> {
            if(line.split(",").length==7){
                return true;
            }else {
                return false;
            }
        });

        JavaRDD<Row> visitActionJavaRDD = filterRDD.map(line -> {
            String[] split = line.split(",");
            String date = split[0];
            String actionTime = split[1];
            String userId = split[2];
            String sessionId = split[3];
            int pageId = Integer.parseInt(split[4].substring(4));
            String clickProductId = split[5];
            String cityId = split[6];
            return RowFactory.create(date, actionTime, userId, sessionId, pageId, clickProductId, cityId);
        });

        StructType schema = DataTypes.createStructType(Arrays.asList(
                DataTypes.createStructField("date", DataTypes.StringType,true),
                DataTypes.createStructField("actionTime", DataTypes.StringType,true),
                DataTypes.createStructField("userId", DataTypes.StringType,true),
                DataTypes.createStructField("sessionId", DataTypes.StringType,true),
                DataTypes.createStructField("pageId", DataTypes.IntegerType,true),
                DataTypes.createStructField("clickProductId", DataTypes.StringType,true),
                DataTypes.createStructField("cityId", DataTypes.StringType,true)
        ));

        Dataset<Row> visitActionDS = sparkSession.createDataFrame(visitActionJavaRDD, schema);

        visitActionDS.registerTempTable("user_visit_action");
        //visitActionDS.show();
        //actionTime|cityId|clickProductId|      date|pageId| sessionId| userId
    }

}
